package streamAPI.sink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.testng.annotations.Test;
import util.FlinkConstant;

/**
 * @author yue.cao
 * @since 10-29-2020
 */
public class DemoWithEasySink {
	@Test
	public void demoWithCSVSink() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.socketTextStream("caoyuesh1", 4444).map(new MapFunction<String, Tuple1<String>>() {
			@Override
			public Tuple1<String> map(String value) throws Exception {
				return Tuple1.of(value);
			}
		}).writeAsCsv("D:/csv", FileSystem.WriteMode.OVERWRITE);
		env.execute();
	}

	@Test
	public void demoWithPrintSink() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Integer> source = env.fromElements(FlinkConstant._Common.FROM_0_TO_10);
		source.addSink(new RichSinkFunction<Integer>() {
			@Override
			public void invoke(Integer value, Context context) throws Exception {
				System.out.println(getRuntimeContext().getIndexOfThisSubtask() + " > " + value);
			}
		});
		env.execute();
	}
}
